package prjCode;

import java.io.IOException;

import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.ec2.model.InstanceType;
import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduce;
import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient;
import com.amazonaws.services.elasticmapreduce.model.HadoopJarStepConfig;
import com.amazonaws.services.elasticmapreduce.model.JobFlowInstancesConfig;
import com.amazonaws.services.elasticmapreduce.model.PlacementType;
import com.amazonaws.services.elasticmapreduce.model.RunJobFlowRequest;
import com.amazonaws.services.elasticmapreduce.model.RunJobFlowResult;
import com.amazonaws.services.elasticmapreduce.model.StepConfig;
import com.amazonaws.services.elasticmapreduce.util.StepFactory;


public class Steps {

	private static final String bucket   = "s3n://dsp-final-project-bucket/"; 
	private static final String assjar   = "s3n://dsp-final-project-bucket/FinalPrj22.jar";
	private static final String output1  = bucket +"output11/";
	private static final String output2  = bucket +"output22/";
	private static final String output3  = bucket +"output33/";
	private static final String output4  = bucket +"output44/";
	private static final String output5  = bucket +"output55/";
	private static final String output6  = bucket +"output66/";
	private static final String output7  = bucket +"output777/";
	private static final String output8  = bucket +"output888/";
	private static final String output9  = bucket +"output999/";
	private static final String output10  = bucket +"output101010/";
	private static final String output11  = bucket +"output111111/";
	private static final String output12  = bucket +"output121212/";
	private static final String output13  = bucket +"output131313/";


	/**
	 * @param args
	 * @param args[0] - DPMinCount
	 * @param args[0] - MinFeatureNum
	 * @throws IOException
	 */
	public static void main(String[] args) throws IOException {
		//AWSCredentials credentials = new PropertiesCredentials(Steps.class.getResourceAsStream("AwsCredentials.properties"));
		AWSCredentials credentials = new BasicAWSCredentials("AKIAJJS7SX5ZANY5FQ6A", "RyDa2svX4+2LFenYWLVOFlHLgjdTAdHgi0TEu2NT");

		AmazonElasticMapReduce mapReduce = new AmazonElasticMapReduceClient(credentials);

		StepConfig debugConfig = new StepConfig().withName("debug")
				.withHadoopJarStep(new StepFactory().newEnableDebuggingStep())
				.withActionOnFailure("TERMINATE_JOB_FLOW");

		HadoopJarStepConfig hadoopJarStep1;
		
		
		hadoopJarStep1 = new HadoopJarStepConfig()
		.withJar(assjar) // This should be a full map reduce application.
		.withMainClass("prjCode.PreprocessingInput")// create paths
		.withArgs("0",output1, args[1]);//0 - run on Amazon, putput1 - output folder, arg[1] - size of test can be 10, 50, 100

		HadoopJarStepConfig hadoopJarStep2 = new HadoopJarStepConfig()
		.withJar(assjar) // This should be a full map reduce application.
		.withMainClass("prjCode.Aggregation")// gather all the results, pass only paths that accrue more then args[2] times
		.withArgs(args[0], output2, args[2]);// size of test, output folder, minimum number of occurrence

		HadoopJarStepConfig hadoopJarStep3 = new HadoopJarStepConfig()
		.withJar(assjar) // This should be a full map reduce application.
		.withMainClass("prjCode.PathFeatureSlotX")//remove path with less features of slot X then args[3]
		.withArgs(output2 , output3, args[3]);//input folder, output folder, minimum number of features
		
		HadoopJarStepConfig hadoopJarStep4 = new HadoopJarStepConfig()
		.withJar(assjar) // This should be a full map reduce application.
		.withMainClass("prjCode.PathFeatureSlotY")//remove path with less features of slot X then args[3]
		.withArgs(output3 , output4, args[3]);//input folder, output folder, minimum number of features
		
		HadoopJarStepConfig hadoopJarStep5 = new HadoopJarStepConfig()
		.withJar(assjar) // This should be a full map reduce application.
		.withMainClass("prjCode.CalcSlotX")
		.withArgs( output4, output5);
		
		HadoopJarStepConfig hadoopJarStep6 = new HadoopJarStepConfig()
		.withJar(assjar) // This should be a full map reduce application.
		.withMainClass("prjCode.CalcSlotY")
		.withArgs( output5, output6);
		
		HadoopJarStepConfig hadoopJarStep7 = new HadoopJarStepConfig()
		.withJar(assjar) // This should be a full map reduce application.
		.withMainClass("prjCode.CalcN")
		.withArgs( output6, output7);
		
		HadoopJarStepConfig hadoopJarStep8 = new HadoopJarStepConfig()
		.withJar(assjar) // This should be a full map reduce application.
		.withMainClass("prjCode.CalcPX")
		.withArgs( output6, output8);
		
		HadoopJarStepConfig hadoopJarStep9 = new HadoopJarStepConfig()
		.withJar(assjar) // This should be a full map reduce application.
		.withMainClass("prjCode.CalcPY")
		.withArgs( output8, output9);
		
		HadoopJarStepConfig hadoopJarStep10 = new HadoopJarStepConfig()
		.withJar(assjar) // This should be a full map reduce application.
		.withMainClass("prjCode.CalcMi")
		.withArgs( output9, output10, output7);
		
		HadoopJarStepConfig hadoopJarStep11 = new HadoopJarStepConfig()
		.withJar(assjar) // This should be a full map reduce application.
		.withMainClass("prjCode.AddIndex")
		.withArgs( output10, output11);
		
		HadoopJarStepConfig hadoopJarStep12 = new HadoopJarStepConfig()
		.withJar(assjar) // This should be a full map reduce application.
		.withMainClass("prjCode.CalcSim")
		.withArgs( output11, output12, args[4]);
		
		HadoopJarStepConfig hadoopJarStep13 = new HadoopJarStepConfig()
		.withJar(assjar) // This should be a full map reduce application.
		.withMainClass("prjCode.OrderResults")
		.withArgs( output12, output13);



		StepConfig stepConfig1 = new StepConfig()
		.withName("pre proccess input")
		.withHadoopJarStep(hadoopJarStep1)
		.withActionOnFailure("TERMINATE_JOB_FLOW");

		StepConfig stepConfig2 = new StepConfig()
		.withName("Sum path occurances")
		.withHadoopJarStep(hadoopJarStep2)
		.withActionOnFailure("TERMINATE_JOB_FLOW");

		StepConfig stepConfig3 = new StepConfig()
		.withName("calc features of Slot X")
		.withHadoopJarStep(hadoopJarStep3)
		.withActionOnFailure("TERMINATE_JOB_FLOW");
		
		StepConfig stepConfig4 = new StepConfig()
		.withName("calc features of Slot Y")
		.withHadoopJarStep(hadoopJarStep4)
		.withActionOnFailure("TERMINATE_JOB_FLOW");
		
		StepConfig stepConfig5 = new StepConfig()
		.withName("calc Slot X affearnce")
		.withHadoopJarStep(hadoopJarStep5)
		.withActionOnFailure("TERMINATE_JOB_FLOW");
		
		StepConfig stepConfig6 = new StepConfig()
		.withName("calc Slot Y affearnce")
		.withHadoopJarStep(hadoopJarStep6)
		.withActionOnFailure("TERMINATE_JOB_FLOW");
		
		StepConfig stepConfig7 = new StepConfig()
		.withName("calc *Slot*")
		.withHadoopJarStep(hadoopJarStep7)
		.withActionOnFailure("TERMINATE_JOB_FLOW");
		
		StepConfig stepConfig8 = new StepConfig()
		.withName("calc PX")
		.withHadoopJarStep(hadoopJarStep8)
		.withActionOnFailure("TERMINATE_JOB_FLOW");
		
		StepConfig stepConfig9 = new StepConfig()
		.withName("calc PY")
		.withHadoopJarStep(hadoopJarStep9)
		.withActionOnFailure("TERMINATE_JOB_FLOW");
		
		StepConfig stepConfig10 = new StepConfig()
		.withName("calc Mi")
		.withHadoopJarStep(hadoopJarStep10)
		.withActionOnFailure("TERMINATE_JOB_FLOW");
		
		StepConfig stepConfig11 = new StepConfig()
		.withName("Add Index")
		.withHadoopJarStep(hadoopJarStep11)
		.withActionOnFailure("TERMINATE_JOB_FLOW");
		
		StepConfig stepConfig12 = new StepConfig()
		.withName("Calc Sim")
		.withHadoopJarStep(hadoopJarStep12)
		.withActionOnFailure("TERMINATE_JOB_FLOW");
		
		StepConfig stepConfig13 = new StepConfig()
		.withName("Order results")
		.withHadoopJarStep(hadoopJarStep13)
		.withActionOnFailure("TERMINATE_JOB_FLOW");


		JobFlowInstancesConfig instances = new JobFlowInstancesConfig()
		.withInstanceCount(8)
		.withMasterInstanceType(InstanceType.M1Small.toString())
		.withSlaveInstanceType(InstanceType.M1Small.toString())
		.withHadoopVersion("0.20").withEc2KeyName("proDspFinal")
		.withKeepJobFlowAliveWhenNoSteps(false)
		.withPlacement(new PlacementType());

		RunJobFlowRequest runFlowRequest = new RunJobFlowRequest()
		.withName("15:18")
		.withInstances(instances)
		.withSteps( debugConfig, stepConfig12, stepConfig13)
		.withLogUri(bucket+"logs/");

		RunJobFlowResult runJobFlowResult = mapReduce.runJobFlow(runFlowRequest);
		String jobFlowId = runJobFlowResult.getJobFlowId();
		System.out.println("Ran job flow with id: " + jobFlowId);
		
		
	}
}
